package com.zallds.window;

import com.zallds.utils.Utils;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;

/**
 * 增量聚合函数(AggregateFunction) +全窗口函数ProcessWindowFunction）结合的综合案例
 * 每经过3秒钟，求安放在北京西站各个红外测温仪所探测到的所有旅客的平均体温。
 */
public class IncrementAndFullAssociationDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.socketTextStream("192.168.62.37", 8889)
                .flatMap(new Utils.Splitter())
                .keyBy(value -> value.f0)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(3)))
                .aggregate(new MyAggregateFunction(), new MyProcessWindowFunction())
                .print();

        env.execute();

    }


    private static class MyAggregateFunction implements   AggregateFunction<Tuple2<String, Integer>, Tuple3<String, Integer, Long>, Tuple2<String, Long>>{
        @Override
        public Tuple3<String, Integer, Long> createAccumulator() {
            return new Tuple3<>("", 0, 0L);
        }

        @Override
        public Tuple3<String, Integer, Long> add(Tuple2<String, Integer> value, Tuple3<String, Integer, Long> tuple3) {
            return new Tuple3<>(value.f0, tuple3.f1 + 1, tuple3.f2 + value.f1);
        }

        @Override
        public Tuple3<String, Integer, Long> merge(Tuple3<String, Integer, Long> acc1, Tuple3<String, Integer, Long> acc2) {
            return new Tuple3<>(acc1.f0, acc1.f1 + acc2.f1, acc1.f2 + acc2.f2);
        }

        @Override
        public Tuple2<String, Long> getResult(Tuple3<String, Integer, Long> tuple3) {
            return new Tuple2<>(tuple3.f0, tuple3.f2 / tuple3.f1);
        }
    }

    private static class MyProcessWindowFunction  extends ProcessWindowFunction<Tuple2<String,Long>,String,String,TimeWindow>{
        @Override
        public void process(String name, Context context, Iterable<Tuple2<String, Long>> elements, Collector<String> out) throws Exception {
            SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            String start = simpleDateFormat.format(context.window().getStart());
            String end = simpleDateFormat.format(context.window().getEnd());

            Long num = 0L;

            for (Tuple2<String, Long> tuple : elements) {
                num = tuple.f1;
            }

            String result = String.format("红外测温仪→ %s, 最近3秒内,测量到的旅客的平均体温是 %s, 窗口的开始时间是→%s,  窗口的结束时间是→%s",name,num,start,end);
            out.collect(result);

        }
    }
}
